package timeandwindow;

/*
* 案例需求：我们这里统计10秒钟的url浏览量，每5秒钟更新一次；
* 另外为了更加清晰地展示，还应该把窗口的起始结束时间一起输出。
* 窗口选择：
*
* */

import com.atguigu.pojo.Event;
import com.atguigu.pojo.UrlViewCount;
import com.atguigu.pojo.WordCount;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.SourceUtil;

import java.time.Duration;

public class Flink10_ReduceFunctionOrAggregateFunctionAndProcessWindowFunction {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);
             //对接数据源，设置水位线
         SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                 .assignTimestampsAndWatermarks(
                         WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );
         ds.print("INPUT");

         //todo 案例需求：统计10秒内每url点击次数, 5秒更新一次，输出包含窗口信息
         //窗口：按键分区事件时间滑动窗口
         ds
                 .map(event -> new WordCount(event.getUrl(),1L))
                 .keyBy(WordCount::getWord)
                 .window(
                         SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))
                 )
                 .aggregate(
                         //增量聚合
                         new AggregateFunction<WordCount, UrlViewCount, UrlViewCount>() {


                             @Override
                             public UrlViewCount createAccumulator() {
                                 return new UrlViewCount(null,null,null,0L);
                             }

                             @Override
                             public UrlViewCount add(WordCount value, UrlViewCount accumulator) {

                                 //将需要输出的多个值封装成对象，方便后面输出
                                 accumulator.setUrl(value.getWord());
                                 accumulator.setCount(accumulator.getCount() + value.getCount());
                                 return accumulator;
                             }

                             @Override
                             public UrlViewCount getResult(UrlViewCount accumulator) {
                                 return accumulator;
                             }

                             @Override
                             public UrlViewCount merge(UrlViewCount urlViewCount, UrlViewCount acc1) {
                                 return null;
                             }
                         },

                         new ProcessWindowFunction<UrlViewCount,UrlViewCount,String,TimeWindow>(){

                             @Override
                             public void process(String key, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context, Iterable<UrlViewCount> elements, Collector<UrlViewCount> out) throws Exception {
                                 //获取增量聚合结果
                                 UrlViewCount urlViewCount = elements.iterator().next();

                                 //获取窗口信息
                                 long windowStart = context.window().getStart();
                                 long windowEnd = context.window().getEnd();

                                 //封装输出结果
                                 urlViewCount.setWindowStart(windowStart);
                                 urlViewCount.setWindowEnd(windowEnd);

                             }
                         }
                 )
                 .print("WINDOW");




         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

}
